package org.apache.rocketmq.broker.processor;

import io.netty.channel.ChannelHandlerContext;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.broker.BrokerController;
import org.apache.rocketmq.broker.metrics.BrokerMetricsManager;
import org.apache.rocketmq.broker.mqtrace.SendMessageContext;
import org.apache.rocketmq.common.AbortProcessException;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.attribute.TopicMessageType;
import org.apache.rocketmq.common.message.*;
import org.apache.rocketmq.common.utils.MessageUtils;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyRemotingAbstract;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.ResponseCode;
import org.apache.rocketmq.remoting.protocol.header.SendMessageRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.SendMessageResponseHeader;
import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingContext;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageStore;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.config.BrokerRole;
import org.apache.rocketmq.store.config.StorePathConfigHelper;

import java.util.Map;
import java.util.Objects;
import java.util.Optional;
import java.util.concurrent.CompletableFuture;

/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2025/6/25
 * @方法描述：Broker模块中处理生产者客户端发送过来的消息的请求处理器
 */
public class SendMessageProcessor extends AbstractSendMessageProcessor implements NettyRequestProcessor {

    public SendMessageProcessor(final BrokerController brokerController) {
        super(brokerController);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理生产者客户端发送过来的消息的方法
     */
    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {

        //定义一个发送消息上下文对象
        SendMessageContext sendMessageContext;
        //根据请求类型执行不同操作
        switch (request.getCode()) {
            //处理消费者消息回退请求
            // case RequestCode.CONSUMER_SEND_MSG_BACK:
            //     return this.consumerSendMsgBack(ctx, request);
            //处理其他类型的消息发送请求
            default:
                //解析请求头
                SendMessageRequestHeader requestHeader = parseRequestHeader(request);
                //如果请求头解析失败，则返回null
                if (requestHeader == null) {
                    return null;
                }
                //构建消息发送上下文对象
                sendMessageContext = buildMsgContext(ctx, requestHeader, request);
                try {
                    //执行消息发送前的回调方法，也就是消息被交给存储引擎之前的回调方法
                    this.executeSendMessageHookBefore(sendMessageContext);
                } catch (AbortProcessException e) {
                    final RemotingCommand errorResponse = RemotingCommand.createResponseCommand(e.getResponseCode(), e.getErrorMessage());
                    //给错误响应设置唯一Id
                    errorResponse.setOpaque(request.getOpaque());
                    return errorResponse;
                }
                //声明响应对象
                RemotingCommand response;
                //清除请求头中的保留属性
                clearReservedProperties(requestHeader);
                //判断要处理的消息是否为批量消息
                if (requestHeader.isBatch()) {
                    //处理批量消息，在第十三版本代码中，我们先不实现消息批处理功能，所以我先把下面的代码注释掉了
                    // response = this.sendBatchMessage(ctx, request, sendMessageContext, requestHeader, mappingContext,
                    //         (ctx1, response1) -> executeSendMessageHookAfter(response1, ctx1));
                    //这是我自己添加的代码
                    response = null;
                } else {
                    //这里就是处理单挑消息的逻辑
                    response = this.sendMessage(ctx, request, sendMessageContext, requestHeader, null,
                            (ctx12, response12) -> executeSendMessageHookAfter(response12, ctx12));
                }
                //返回处理响应
                return response;
        }
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：把消息交给消息存储引擎处理的核心方法
     */
    public RemotingCommand sendMessage(final ChannelHandlerContext ctx,
                                       final RemotingCommand request,
                                       final SendMessageContext sendMessageContext,
                                       final SendMessageRequestHeader requestHeader,
                                       final TopicQueueMappingContext mappingContext,
                                       final SendMessageCallback sendMessageCallback) throws RemotingCommandException {

        //执行消息预处理操作，所谓预处理其实就是在消息发送之前对消息进行一些校验，比如权限检查、主题验证等
        final RemotingCommand response = preSend(ctx, request, requestHeader);
        //如果预处理操作执行失败，直接返回错误响应
        if (response.getCode() != -1) {
            return response;
        }
        //得到响应头对象
        final SendMessageResponseHeader responseHeader = (SendMessageResponseHeader) response.readCustomHeader();
        //从请求中得到要处理的消息体
        final byte[] body = request.getBody();
        //得到消息要发送的目标队列
        int queueIdInt = requestHeader.getQueueId();
        //获取主题配置信息对象
        TopicConfig topicConfig = this.brokerController.getTopicConfigManager().selectTopicConfig(requestHeader.getTopic());
        //如果队列Id无效，则随机选择一个写队列
        if (queueIdInt < 0) {
            queueIdInt = randomQueueId(topicConfig.getWriteQueueNums());
        }
        //创建内部消息对象，这个消息对象就会被传递给消息存储引擎
        MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
        //设置消息主题
        msgInner.setTopic(requestHeader.getTopic());
        //设置目标队列Id
        msgInner.setQueueId(queueIdInt);
        //将消息属性字符串转换为Map
        Map<String, String> oriProps = MessageDecoder.string2messageProperties(requestHeader.getProperties());
        //处理重试和死信队列逻辑，但现在我们肯定还有实现这些功能，所以我先把代码注释了
        // if (!handleRetryAndDLQ(requestHeader, response, request, msgInner, topicConfig, oriProps)) {
        //     return response;
        // }
        //设置消息体
        msgInner.setBody(body);
        //设置消息标志
        msgInner.setFlag(requestHeader.getFlag());
        //得到消息的唯一Id
        String uniqKey = oriProps.get(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX);
        //如果消息没有唯一Id，那就创建一个
        if (uniqKey == null || uniqKey.length() <= 0) {
            uniqKey = MessageClientIDSetter.createUniqID();
            oriProps.put(MessageConst.PROPERTY_UNIQ_CLIENT_MESSAGE_ID_KEYIDX, uniqKey);
        }
        //设置消息属性
        MessageAccessor.setProperties(msgInner, oriProps);
        //将消息的过滤标签转换为标签码
        msgInner.setTagsCode(MessageExtBrokerInner.tagsString2tagsCode(topicConfig.getTopicFilterType(), msgInner.getTags()));
        //设置消息创建时间戳
        msgInner.setBornTimestamp(requestHeader.getBornTimestamp());
        //设置消息来源客户端的网络地址
        msgInner.setBornHost(ctx.channel().remoteAddress());
        //设置消息存储到的服务器的网络地址
        msgInner.setStoreHost(this.getStoreHost());
        //设置消息重试次数
        msgInner.setReconsumeTimes(requestHeader.getReconsumeTimes() == null ? 0 : requestHeader.getReconsumeTimes());
        //得到当前broker节点所在集群名称
        String clusterName = this.brokerController.getBrokerConfig().getBrokerClusterName();
        //设置集群名称
        MessageAccessor.putProperty(msgInner, MessageConst.PROPERTY_CLUSTER, clusterName);
        //将消息属性转换为字符串格式
        msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgInner.getProperties()));
        //得到消息存储的开始时间
        long beginTimeMillis = this.brokerController.getMessageStore().now();
        //判断是否启用异步存储
        if (brokerController.getBrokerConfig().isAsyncSendEnable()) {
            //如果启用了异步存储，那就在这里把消息交给存储引擎去异步存储
            CompletableFuture<PutMessageResult> asyncPutMessageFuture = this.brokerController.getMessageStore().asyncPutMessage(msgInner);
            //记录消息发送的目标队列Id
            final int finalQueueIdInt = queueIdInt;
            //记录消息本身
            final MessageExtBrokerInner finalMsgInner = msgInner;
            //向future对象中注册回调方法，该方法会处理消息存储的结果
            asyncPutMessageFuture.thenAcceptAsync(putMessageResult -> {
                //处理消息存储结果，并且生成要回复给proxy模块的响应对象，然后把响应回复给proxy客户端
                RemotingCommand responseFuture =
                        handlePutMessageResult(putMessageResult, response, request, finalMsgInner, responseHeader, sendMessageContext,
                                ctx, finalQueueIdInt, beginTimeMillis, mappingContext, BrokerMetricsManager.getMessageType(requestHeader));
                //如果有响应对象，则回复给proxy客户端
                if (responseFuture != null) {
                    //这里并没有得到响应，所以不会进入下面的分支
                    doResponse(ctx, request, responseFuture);
                }
                //执行发送消息完成之后的回调方法
                sendMessageCallback.onComplete(sendMessageContext, response);
                //设置执行回调方法的执行器
            }, this.brokerController.getPutMessageFutureExecutor());
            return null;
        }
        else {
            //执行到这里就意味着是同步发送消息操作，在这里把消息交给存储引擎存储，同步等待结果返回
            PutMessageResult putMessageResult = this.brokerController.getMessageStore().putMessage(msgInner);
            //处理消息存储结果，并设置响应对象
            handlePutMessageResult(putMessageResult, response, request, msgInner, responseHeader, sendMessageContext, ctx, queueIdInt, beginTimeMillis, mappingContext, BrokerMetricsManager.getMessageType(requestHeader));
            //执行发送消息完成之后的回调方法
            sendMessageCallback.onComplete(sendMessageContext, response);
            //返回响应对象
            return response;
        }
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：是否拒绝处理请求的方法
     */
    @Override
    public boolean rejectRequest() {
        //如果当前Broker节点没有开始代理主机点模式，并且当前节点并不是主节点，则拒绝处理请求
        //因为只有broker主节点才能处理消息，从节点只是用作数据备份
        if (!this.brokerController.getBrokerConfig().isEnableSlaveActingMaster() && this.brokerController.getMessageStoreConfig().getBrokerRole() == BrokerRole.SLAVE) {
            return true;
        }
        //如果操作系统缓存页资源不足或者消息存储引擎没有开启临时缓冲池功能，或者开启了但是临时缓冲池资源不足，那就拒绝处理该消息
        if (this.brokerController.getMessageStore().isOSPageCacheBusy() || this.brokerController.getMessageStore().isTransientStorePoolDeficient()) {
            return true;
        }

        return false;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：处理消息存储结果的方法，在该方法中，消息存储结果会被封装到RemotingCommand响应对象中，该方法在第十三版本代码中还不需要添加注释
     * 等后面实现回复客户端响应功能时再重构该方法，添加详细注释
     */
    private RemotingCommand handlePutMessageResult(PutMessageResult putMessageResult, RemotingCommand response,
                                                   RemotingCommand request, MessageExt msg, SendMessageResponseHeader responseHeader,
                                                   SendMessageContext sendMessageContext, ChannelHandlerContext ctx, int queueIdInt, long beginTimeMillis,
                                                   TopicQueueMappingContext mappingContext, TopicMessageType messageType) {

        boolean sendOK = false;

        switch (putMessageResult.getPutMessageStatus()) {
            // Success
            case PUT_OK:
                sendOK = true;
                response.setCode(ResponseCode.SUCCESS);
                break;
            case FLUSH_DISK_TIMEOUT:
                response.setCode(ResponseCode.FLUSH_DISK_TIMEOUT);
                sendOK = true;
                break;
            case FLUSH_SLAVE_TIMEOUT:
                response.setCode(ResponseCode.FLUSH_SLAVE_TIMEOUT);
                sendOK = true;
                break;
            case SLAVE_NOT_AVAILABLE:
                response.setCode(ResponseCode.SLAVE_NOT_AVAILABLE);
                sendOK = true;
                break;

            // Failed
            case IN_SYNC_REPLICAS_NOT_ENOUGH:
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("in-sync replicas not enough");
                break;
            case CREATE_MAPPED_FILE_FAILED:
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("create mapped file failed, server is busy or broken.");
                break;
            case MESSAGE_ILLEGAL:
            case PROPERTIES_SIZE_EXCEEDED:
                response.setCode(ResponseCode.MESSAGE_ILLEGAL);
                response.setRemark(String.format("the message is illegal, maybe msg body or properties length not matched. msg body length limit %dB, msg properties length limit 32KB.",
                        this.brokerController.getMessageStoreConfig().getMaxMessageSize()));
                break;
            case WHEEL_TIMER_MSG_ILLEGAL:
                response.setCode(ResponseCode.MESSAGE_ILLEGAL);
                response.setRemark(String.format("timer message illegal, the delay time should not be bigger than the max delay %dms; or if set del msg, the delay time should be bigger than the current time",
                        this.brokerController.getMessageStoreConfig().getTimerMaxDelaySec() * 1000L));
                break;
            case WHEEL_TIMER_FLOW_CONTROL:
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark(String.format("timer message is under flow control, max num limit is %d or the current value is greater than %d and less than %d, trigger random flow control",
                        this.brokerController.getMessageStoreConfig().getTimerCongestNumEachSlot() * 2L, this.brokerController.getMessageStoreConfig().getTimerCongestNumEachSlot(), this.brokerController.getMessageStoreConfig().getTimerCongestNumEachSlot() * 2L));
                break;
            case WHEEL_TIMER_NOT_ENABLE:
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark(String.format("accurate timer message is not enabled, timerWheelEnable is %s",
                        this.brokerController.getMessageStoreConfig().isTimerWheelEnable()));
                break;
            case SERVICE_NOT_AVAILABLE:
                response.setCode(ResponseCode.SERVICE_NOT_AVAILABLE);
                response.setRemark(
                        "service not available now. It may be caused by one of the following reasons: " +
                                "the broker's disk is full [" + diskUtil() + "], messages are put to the slave, message store has been shut down, etc.");
                break;
            case OS_PAGE_CACHE_BUSY:
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("[PC_SYNCHRONIZED]broker busy, start flow control for a while");
                break;
            case LMQ_CONSUME_QUEUE_NUM_EXCEEDED:
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("[LMQ_CONSUME_QUEUE_NUM_EXCEEDED]broker config enableLmq and enableMultiDispatch, lmq consumeQueue num exceed maxLmqConsumeQueueNum config num, default limit 2w.");
                break;
            case UNKNOWN_ERROR:
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("UNKNOWN_ERROR");
                break;
            default:
                response.setCode(ResponseCode.SYSTEM_ERROR);
                response.setRemark("UNKNOWN_ERROR DEFAULT");
                break;
        }

        response.setRemark(null);
        //这里暂且注释掉，因为响应信息还不完善，不注释的话有空指针异常，无法把响应顺利回复给生产者客户端
        //responseHeader.setMsgId(putMessageResult.getAppendMessageResult().getMsgId());
        //responseHeader.setQueueId(queueIdInt);
        //responseHeader.setQueueOffset(putMessageResult.getAppendMessageResult().getLogicsOffset());
        //responseHeader.setTransactionId(MessageClientIDSetter.getUniqID(msg));
        doResponse(ctx, request, response);
        return response;
    }


    private void clearReservedProperties(SendMessageRequestHeader requestHeader) {
        String properties = requestHeader.getProperties();
        properties = MessageUtils.deleteProperty(properties, MessageConst.PROPERTY_POP_CK);
        requestHeader.setProperties(properties);
    }

    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：统计磁盘利用率的方法
     */
    private String diskUtil() {
        double physicRatio = 100;
        String storePath;
        MessageStore messageStore = this.brokerController.getMessageStore();
        if (messageStore instanceof DefaultMessageStore) {
            storePath = ((DefaultMessageStore) messageStore).getStorePathPhysic();
        } else {
            storePath = this.brokerController.getMessageStoreConfig().getStorePathCommitLog();
        }
        String[] paths = storePath.trim().split(MixAll.MULTI_PATH_SPLITTER);
        for (String storePathPhysic : paths) {
            physicRatio = Math.min(physicRatio, UtilAll.getDiskPartitionSpaceUsedPercent(storePathPhysic));
        }

        String storePathLogis =
                StorePathConfigHelper.getStorePathConsumeQueue(this.brokerController.getMessageStoreConfig().getStorePathRootDir());
        double logisRatio = UtilAll.getDiskPartitionSpaceUsedPercent(storePathLogis);

        String storePathIndex =
                StorePathConfigHelper.getStorePathIndex(this.brokerController.getMessageStoreConfig().getStorePathRootDir());
        double indexRatio = UtilAll.getDiskPartitionSpaceUsedPercent(storePathIndex);

        return String.format("CL: %5.2f CQ: %5.2f INDEX: %5.2f", physicRatio, logisRatio, indexRatio);
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @方法描述：消息预处理方法
     */
    private RemotingCommand preSend(ChannelHandlerContext ctx, RemotingCommand request,
                                    SendMessageRequestHeader requestHeader) {
        //创建一个返回给proxy节点的响应对象，响应对象的响应头为SendMessageResponseHeader
        final RemotingCommand response = RemotingCommand.createResponseCommand(SendMessageResponseHeader.class);
        //设置响应对象的唯一Id，用来和请求一一对应
        response.setOpaque(request.getOpaque());
        //给响应对象设置额外的属性，设置消息区域Id
        response.addExtField(MessageConst.PROPERTY_MSG_REGION, this.brokerController.getBrokerConfig().getRegionId());
        //设置是否跟踪消息的标志
        response.addExtField(MessageConst.PROPERTY_TRACE_SWITCH, String.valueOf(this.brokerController.getBrokerConfig().isTraceOn()));
        LOGGER.debug("Receive SendMessage request command {}", request);
        //得到broker配置信息中设定的开始接受发送请求的时间戳
        //这个时间戳用于控制broker的启动预热期，在此时间之前不会处理生产者客户端发送过来的消息
        final long startTimestamp = this.brokerController.getBrokerConfig().getStartAcceptSendRequestTimeStamp();
        //检查当前时间是否小于设定的开始接收生产者消息的时间戳，如果当前时间早于设定时间，表示broker尚未准备好接收消息
        if (this.brokerController.getMessageStore().now() < startTimestamp) {
            //设置响应码为系统错误
            response.setCode(ResponseCode.SYSTEM_ERROR);
            //设置响应备注信息，表示broker无法提供服务的原因和可提供服务的时间
            response.setRemark(String.format("broker unable to service, until %s", UtilAll.timeMillisToHumanString2(startTimestamp)));
            //返回响应对象
            return response;
        }
        //执行到这里意味着当前broker节点可以接收生产者消息了
        //但是还没有真的处理请求，所以先把响应码预设成-1，后面处理完之后再更新响应码
        response.setCode(-1);
        //调用父类的消息检查方法，对消息进行一系列校验
        super.msgCheck(ctx, requestHeader, request, response);
        return response;
    }


}
